/**
 * Copyright 2009 Takahiko Ito
 *
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *    http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.unigram.oluolu.rqe.postprocessor;

import java.io.BufferedReader;
import java.io.FileInputStream;
import java.io.IOException;
import java.io.InputStreamReader;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.Map;
import java.util.Set;
import java.util.Vector;

import org.apache.hadoop.conf.Configuration;
import org.unigram.oluolu.common.OluoluConstants;

/**
 * An implementation of IPostProcessor. This class is extended 
 * NgramPostProcessor. This object validates whether input related 
 * query pairs are mistaken query and the correction query.  
 */
public class SpellCheckPostProcessor
extends NgramPostProcessor {

    /**
     * Constructor.
     * 
     * @param conf configuration
     */
    public SpellCheckPostProcessor(final Configuration conf) {
        super(conf);
        this.readingDictionary = new HashMap<String, Set<String>>();
        this.likelihoodThredhold = conf.getFloat(LIKELIHOOD_THRESHOLD, 
                DEFAULT_LIKELIHOOD_THRESHOLD);
        this.levenshteinThreshold = conf.getLong(RQE_LEVENSHTEIN_THRESHOLD,
                DEFAULT_RQE_LEVENSHTEIN_THRESHOLD);
        this.lengthDiffThreshold = conf.getLong(RQE_LENGTH_DIFF_THRESHOLD,
                DEFAULT_RQE_LENGTH_DIFF_THRESHOLD);        
        this.loadDictionary(conf.get(OluoluConstants.LANGUAGE, "us"));        
    }

    /**
     * Loading reading dictionary file and create the map
     * (key: character, value readings).  
     * 
     * @param lang language of input data.
     */
    private void loadDictionary(final String lang) {
        String readingDictionaryFile = 
            this.READING_DICTIONAY_DIRECTORY + "/char_reading_default.dic";
        if (lang.equals("ja")) {
            readingDictionaryFile = 
                this.READING_DICTIONAY_DIRECTORY + "/char_reading_jp.dic";
        } 
        try{
            InputStreamReader isr = new InputStreamReader(
                    new FileInputStream(readingDictionaryFile), "utf-8");
            BufferedReader br = new BufferedReader(isr);
            String tmpStr;
            String[] segs;
            while((tmpStr = br.readLine()) != null){
                segs = tmpStr.split("\t");
                Set<String> readings = new HashSet<String>();
                for (int i=1; i < segs.length; i++) {
                    readings.add(segs[i]);
                }
                this.readingDictionary.put(segs[0], readings);
            }
            br.close();
        }catch(IOException e){
            e.printStackTrace();
        }
    }

    /**
     * Validate whether the query pair should be added to dictionary.
     * 
     * @param firstFreq frequency of first query
     * @param secondFreq  frequency of second query
     * @param pairFreq frequency of the occurrence of first query and 
     *         second query in the same sessions
     * @param totalFreq the number of query in the input data
     * @return true when the query is added to a dictionary 
     *          otherwise returns false
     */
    @Override
    protected final boolean validate(final Long firstFreq,
            final Long secondFreq, final Long pairFreq, 
            final Long totalFreq) {
        if (firstFreq > secondFreq) {
            return false;
        }
        float ratio = likelihoodRatio(firstFreq, secondFreq, 
                pairFreq, totalFreq);
        if (ratio  > this.likelihoodThredhold) {
            return true;
        } 
        return false;
    }

    /**
     * Validate the pair of queries whether they are related or not based
     * on their string similarity.
     * 
     * @param firstQuery query string
     * @param secondQuery  query string
     * @return true when they are related 
     */
    @Override
    protected boolean validateByStrings(final String firstQuery, 
            final String secondQuery) {
        
        if (firstQuery ==null || secondQuery == null) {
            return false;
        }        
        if (firstQuery.length() > 10 || secondQuery.length() > 10) {
            return false;
        }        
        if (firstQuery.length() < 2 || secondQuery.length() < 2) {
            return false;
        }

        Vector<String> firstCandidates  = this.generateCandidates(firstQuery);
        Vector<String> secondCandidates = this.generateCandidates(secondQuery);
        
        int maxCandidate = 100;
        int firstCandidateSize = (firstCandidates.size() > maxCandidate
                    ? maxCandidate : firstCandidates.size());
        int secondCandidateSize = (secondCandidates.size() > maxCandidate
                    ? maxCandidate : secondCandidates.size());

        for (int i=0; i<firstCandidateSize; i++) {
            for (int j=0; j<secondCandidateSize; j++) {
                int result = this.calcSimilarity(
                        firstCandidates.get(i).toString(), 
                        secondCandidates.get(j).toString());
                if (result < this.levenshteinThreshold) {
                    return true;
                }
            }
        }
        return false;
    }  
    
    /**
     * Compute the similarity between two input query string by edit distance. 
     * @param firstQuery query string
     * @param secondQuery query string
     * @return distance between two query strings.
     */
    private int calcSimilarity(final String firstQuery, 
            final String secondQuery) {

        /* calc edit distance */
        int[][] d = new int[firstQuery.length() + 1][secondQuery.length() + 1];
        for (int i = 0; i <= firstQuery.length(); i++) {
            d[i][0] = i; // deletion 
        }
        for (int j = 0; j <= secondQuery.length(); j++) {
            d[0][j] = j; // insertion
        }

        for (int j = 0; j < secondQuery.length(); j++) {
            for (int i = 0; i < firstQuery.length(); i++) {
                if (firstQuery.charAt(i) == secondQuery.charAt(j)) {
                    d[i + 1][j + 1] = d[i][j];
                } else {
                    d[i + 1][j + 1] = Math
                    .min(d[i][j + 1] + 1, d[i + 1][j] + 1);
                    d[i + 1][j + 1] = Math.min(d[i + 1][j + 1], d[i][j] + 1);
                }
            }
        }
        return d[firstQuery.length()][secondQuery.length()];
    }

    /**
     * Generate reading candidates.
     * @param queryStr query string
     * @return a set of candidate readings of input query string.
     */
    private Vector<String> generateCandidates(final String queryStr){
        Vector<String> prevReadings = new Vector<String>();
        Vector<String> currentReadings = new Vector<String>();
        for (int i=0; i< queryStr.length(); i++) {
            currentReadings = new Vector<String>();    		
            String tmpChar = String.valueOf(queryStr.charAt(i));
            Set<String> readings = this.readingDictionary.get(tmpChar);
            if (readings == null) { // no reading in dictionary 
                if (i == 0) {
                    currentReadings.add(tmpChar);        			
                } else { // 
                    Iterator<String> prevIter = prevReadings.iterator();
                    while (prevIter.hasNext()) {
                        String prevReading = prevIter.next();
                        currentReadings.add(prevReading + tmpChar);
                    }
                }
                continue;
            }
            Iterator<String> readingIter = readings.iterator();
            while (readingIter.hasNext()) {
                String reading = readingIter.next();
                //System.err.println("     reading: " + reading);
                if (i == 0) {
                    currentReadings.add(reading);
                } else {
                    Iterator<String> prevIter = prevReadings.iterator();
                    if (prevReadings.size() > 100) { // not generate too much
                        continue;
                    }
                    while (prevIter.hasNext()) {
                            String prevReading = prevIter.next();
                            currentReadings.add(prevReading + reading);
                    }
                }
            }
            prevReadings = currentReadings;
        }
        return currentReadings;
    }

    // TODO create test cases
    /**
     * Validate query pair based on log likelihood ratio.
     * 
     * @param firstFreq frequency of first query
     * @param secondFreq  frequency of second query
     * @param pairFreq frequency of the occurrence of first query and 
     *         second query in the same sessions
     * @param totalFreq the number of query in the input data
     * @return log likelihood ratio
     */
    private static float likelihoodRatio(final Long firstFreq,
            final Long secondFreq, final Long pairFreq, 
            final Long totalFreq) {
        float p  = (float) secondFreq / (float) totalFreq;
        float p1 = (float) pairFreq / (float) firstFreq;
        float p2 = (float) (secondFreq - pairFreq) 
        / (float) (totalFreq - firstFreq);
        float l1 = (float) Math.log(likelihood(pairFreq, firstFreq, p));
        float l2 = (float) Math.log(likelihood((secondFreq-pairFreq), 
                (totalFreq - firstFreq), p));
        float l3 = (float) Math.log(likelihood(pairFreq, firstFreq, p1));
        float l4 = (float) Math.log(likelihood((secondFreq - pairFreq), 
                (totalFreq - firstFreq), p2));
        return -2 * (l1 + l2 - l3 - l4); 
    }

    /**
     * Calculate likelihood.
     * 
     * @param k the number of occurrence
     * @param n the number of try
     * @param x probability
     * @return likelihood
     */
    private static double likelihood(final long k, 
            final long n, final double x) {
        double result = 1.0;
        return Math.pow(x, k) * Math.pow((1-x), (n-k));
    }

    /** threshold of likelihood ratio. */
    private final float likelihoodThredhold;

    /** symbol: likelihood threshold. */
    public static final String LIKELIHOOD_THRESHOLD
    = "oluolu.rqe.likelihood.threshold";    

    /** directory. */
    private static final String READING_DICTIONAY_DIRECTORY	
    = "conf/dic/spell";    

    /** default: threshold for likelihood rate. */
    public static final float DEFAULT_LIKELIHOOD_THRESHOLD = 1.5F;

    /** reading dictionary map. */
    private Map<String, Set<String>> readingDictionary;

    /** symbol: levenshtein threshold. */
    public static final String RQE_LEVENSHTEIN_THRESHOLD
    = "oluolu.rqe.levenshtein.threshold";

    /** default: threshold for levenshtein distance. */
    public static final long DEFAULT_RQE_LEVENSHTEIN_THRESHOLD = 3;    

    /** symbol: length difference threshold. */
    public static final String RQE_LENGTH_DIFF_THRESHOLD
    = "oluolu.rqe.length.diff.threshold";

    /** default: threshold for length difference. */
    public static final int DEFAULT_RQE_LENGTH_DIFF_THRESHOLD = 3;    

    /** edit distance threshold. */
    private long levenshteinThreshold;

    /** length difference threshold. */
    private long lengthDiffThreshold;

}
